package org.ala.tiktools.rpc.kcp.protocol;

import java.util.List;

import org.ala.tiktools.rpc.kcp.config.RpcKcpConfigInfo;
import org.ala.tiktools.rpc.kcp.protocol.receiver.AckList;
import org.ala.tiktools.rpc.kcp.protocol.receiver.KcpReceiver2;
import org.ala.tiktools.rpc.kcp.protocol.segment.KcpSegment;
import org.ala.tiktools.rpc.kcp.protocol.sender.KcpSender2;
import org.ala.tiktools.rpc.kcp.protocol.tools.KcpCongestion;
import org.ala.tiktools.rpc.kcp.protocol.tools.KcpMetric;
import org.ala.tiktools.rpc.kcp.protocol.tools.KcpProbe;
import org.ala.tiktools.rpc.kcp.protocol.tools.KcpRto;
import org.ala.tiktools.tools.Assert;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.channel.Channel;


/**
 * kcp协议实现
 * <p>	send：上层应用把待发送的消息写入sndQue，等待flush时发送出去（上层应用就是netty层）
 * <p>	update：按时间戳触发flush
 * <p>	flush：把sndBuf中的数据调用IKcpOutput发送出去（IKcpOutput对应boss线程的udp socket），然后从sndQue读走消息到sndBuf，等待下个flush
 * <p>			sndQue -> sndBuf 时按rmtWnd（远程窗口大小）为上限，写走了多少，抽进来多少
 * <p>			写走的消息中（KCP头）中，告诉对方自己的 rcvWnd 还能容纳多少数据（就是对方的rmtWnd） 
 * 
 * <p>	input：上层应用从rcvQue中读走消息，然后rcvQue再从rcvBuf读走消息等待下个input（上层应用就是netty层）
 * 				rcvBuf -> rcvQue 时按rcvWmd（本地窗口大小）为上限，读走了多少，抽进来多少
 * <p>	recv：把udp socket 读到的数据，读进rcvBuf
 * 
 * 
 * @author ala
 * @date 2025-04-11 11:20
 */
public class Kcp implements IKcp {


    /**
     *  绑定的channel
     *  <p> KcpClientChannel
     */
    protected Channel channel;
    /**
     *  远程身份标识
     */
    protected int conv;
    /**
     *  最大包长
     */
    protected int mtu;
    /**
     *  最大数据段长（mtu - 24）
     */
    protected int mss;
    /**
     *  当前状态
     */
    protected int state;
    /**
     *  刷新间隔
     */
    protected int interval;
    /**
     *  fastResend
     *  <p>	默认不开启
     */
    protected int fastResend = -1;
    /**
     *  允许的快速重试次数，超过此数则不再快速重试
     *  <p> <=0 表示不限
     *  <p> 超过此数的seg，将走普通reSendTs时间机制进行重试
     */
    protected int fastLimit;
    /**
     *  最大重试次数，超过此次数kcp状态会置为-1
     */
    protected int deadLink;
    /**
     *  流模式
     *  <p> 二进制流会尽量塞满一个mss再发送。但这样会带俩粘包问题
     */
    protected boolean stream;


    /**
     *  update标记
     */
    protected boolean updated;
    /**
     *  上次刷新时间戳
     */
    protected int tsFlush;
    /**
     *  远程窗口大小
     *  <p> 实时变化的，对方还能接收的数据段数
     */
    protected int rmtWnd;
    /**
     *  当前时间戳
     *  <p> 纳秒时间戳转为毫秒数，再取后32位
     */
    protected int current;
    /**
     *  自动conv模式
     *  <p> 以第一个收到的消息中的conv，为本kcp的conv
     */
    protected boolean autoSetConv;
    /**
     *
     */
    protected ByteBufAllocator byteBufAllocator = ByteBufAllocator.DEFAULT;


    public Kcp(int conv, IKcpOutput output) {
        this.conv = conv;
        this.kcpName = String.format("kcp:{conv:%d}", conv);
        
        interval = RpcKcpConfigInfo.INSTANCE.getiKcpInterval();
        mtu = RpcKcpConfigInfo.INSTANCE.getiKcpMtuDef();
        mss = mtu - IKCP_OVERHEAD;
        fastLimit = RpcKcpConfigInfo.INSTANCE.getiKcpFastAckLimit();
        deadLink = RpcKcpConfigInfo.INSTANCE.getiKcpDeadLink();

        //  初始化receiver
        receiver = new KcpReceiver2(this, RpcKcpConfigInfo.INSTANCE.getiKcpWndRcv());

        //  初始化sender
        sender = new KcpSender2(this, RpcKcpConfigInfo.INSTANCE.getiKcpWndSnd(), output);
        //  初始化ackList
        ackList = new AckList(8);
        //  初始化拥塞窗口
        congestion = new KcpCongestion(this, 1, mss, RpcKcpConfigInfo.INSTANCE.getiKcpThreshInit(), mss);

        //  初始化rto
        rto = new KcpRto(
                this,
                RpcKcpConfigInfo.INSTANCE.getiKcpRtoDef(),
                RpcKcpConfigInfo.INSTANCE.getiKcpRtoMin(),
                RpcKcpConfigInfo.INSTANCE.isiKcpNoDelay());
        //  初始化probe
        probe = new KcpProbe(this);
        //  初始化metric
        metric = new KcpMetric(this);
    }


    /************************************************************************************************************************
     *  读相关
     ************************************************************************************************************************/
    /**
     *
     */
    protected KcpReceiver2 receiver;
    /**
     * rcvQue -> buf
     */
    @Override
    public int recv(ByteBuf buf) {
        long tag = receiver.recv(buf);
        int probe = (int)(tag >>> 32);
        int len = (int)tag;
        this.probe.probe(probe);
        return len;
    }
    /**
     *  rcvQue[i].data -> buf
     */
    @Override
    public int recv(List<ByteBuf> bufs) {
        long tag = receiver.recv(bufs);
        int probe = (int)(tag >>> 32);
        int len = (int)tag;
        this.probe.probe(probe);
        return len;
    }
    /**
     *  io buf -> rcvBuf
     *  <p> 连续的读，直到buf中不够1个kcp头为止
     *
     *  @return -1:buf中的数据本来就不够读一个kcp头（1个都没读的还）<br/>
     *          -2:非法数据（可读区域不够1个数据段，或者数据段长度 < 0）<br/>
     *          -3:非法消息（cmd不可识别）<br/>
     *          -4:conv不是发给本conv的（收到了别人的数据）<br/>
     *          0:接收成功
     */
    @Override
    public int input(ByteBuf buf) {
        //  接收到的数据不够1个kcp头
        if (buf == null || buf.readableBytes() < IKcp.IKCP_OVERHEAD) {
            return -1;
        }
        //  记录本次input收到的最大ack的sn
        long maxAckSn = -1;
        //  读之前的sndUna（如果发生了改变，需要更新拥塞窗口）
        long oldSndUna = sender.getSndUna();

        //  接收消息
        long tag = receiver.input(buf);
        int probe = (int)(tag >>> 32);
        int res = (int) tag;
        if (res < 0) { return res; }
        maxAckSn = res;
        this.probe.probe(probe);

        //  如果本次收到了ack，根据ack的最大sn号，累加最大ackSn之前的fastAck
        sender.parseFastack(maxAckSn);

        //  如果sndUna往前递推了，重新计算拥塞窗口
        //  rmtWnd在 receiver.input 过程中可能会更新
        congestion.updateCWnd(oldSndUna, rmtWnd);
        return 0;
    }


    /************************************************************************************************************************
     *  写相关
     ************************************************************************************************************************/
    /**
     *
     */
    protected KcpSender2 sender;
    /**
     *  待回复列表
     */
    protected AckList ackList;
    /**
     *  发送拥塞窗口
     */
    protected KcpCongestion congestion;
    /**
     * buf中的数据 -> sndQue
     * @return 写入的字节数
     */
    public int send(ByteBuf buf) {
        Assert.assertTrue(mss > 0, "数据段长(mss = mtu - 24)不能=0");
        if (buf.readableBytes() == 0) { return -1; }
        return sender.sendToSndQue(buf);
    }
    /**
     *  flush方法（实际发送数据）
     */
    public void flush() {
        if (!updated) { return; }
        //  写入的buf，各个写入方法内部会初始化
        ByteBuf buf = null;
        //  写入的seg模版（避免反复gc）
        KcpSegment seg = KcpSegment.alloc(byteBufAllocator, 0);
        seg.setConv(conv);
        seg.setFrg((short) 0);
        seg.setWnd(receiver.wndUnused());
        seg.setUna(receiver.getRcvNxt());
        seg.setSn(0);
        seg.setTs(0);
        //  把ackList中的数据发送出去先
        buf = sender.writeAndClearAckList(ackList, buf, seg);

        //  询问窗口大小，或回答窗口大小
        probe.probeAskTell();
        if (probe.isAskWnd()) {
            buf = sender.writeAskWnd(buf, seg);
        }
        if (probe.isTellWnd()) {
            buf = sender.writeTellWnd(buf, seg);
        }
        //  清空probe
        probe.clear();

        //  返回当前的拥塞窗口，如果算上拥塞窗口，还有发送容量，则sndQue -> sndBuf 准备发送（sndNxt < sndUna + cwnd）
        int cwnd = congestion.getCwnd();
        
        //  sndQue -> sndBuf
        sender.moveQueToBuf(cwnd, seg.getWnd());
        //  发送sndBuf
        buf = sender.sendSndBuf(buf, seg.getWnd());
        //  把尾巴发送出去
        if (buf != null) {
            if (buf.readableBytes() > 0) {
                sender.output(buf);
            } else {
                buf.release();
            }
        }
    }
    /**
     *  update方法，尝试把消息发送出去，并计算下次触发时间戳
     *  <p>	下次触发时间戳：max(tsUpdate, currrent) + 100ms
     */
    public void update(int current) {
        this.current = current;
        //  如果是第一次update，直接
        if (!updated) {
            updated = true;
            tsFlush = current;
        }
        //  计算离上次flush的毫秒数
        int slap = current - tsFlush;
        //  如果已经超过10s则直接刷新（slag == 0）
        if (slap >= 10000 || slap <= -10000) {
            tsFlush = current;
            slap = 0;
        }
        if (slap >= 0) {
            //  如果 current 超过了下个刷新时间戳，tsFlush直接置为下个刷新时间戳
            tsFlush += interval;
            if (current > tsFlush) {
                tsFlush = current + interval;
            }
        } else {
            tsFlush = current + interval;
        }
        //  执行flush
        flush();
    }
    /**
     *  检测current
     */
    public int check(int current) {
        if (!updated) {return current;}
        int tsFlush = this.tsFlush;
        int slap = current - tsFlush;
        if (slap >= 10000 || slap < -10000) {
            tsFlush = current;
            slap = 0;
        }
        if (slap >= 0) { return current; }

        return sender.checkSndBufReSendTs(current, tsFlush);
    }
    /**
     * 检测当前是否可flush()
     * <p>  1:ackList不为空（有ack需要回复）<br/>
     *      2:probe不为0（有窗口询问或窗口询问需要回答）<br/>
     *      3:sndBuf不为空，sndQue不为空<br/>
     */
    public boolean checkFlush() {
        if (ackList.size() > 0) {
            return true;
        }
        if (probe.getProbe() != 0) {
            return true;
        }
        if (!sender.getSndBuf().isEmpty()) {
            return true;
        }
        if (!sender.getSndQue().isEmpty()) {
            return true;
        }
        return false;
    }


    /************************************************************************************************************************
     *  其他相关
     ************************************************************************************************************************/
    /**
     *  rto计算相关
     */
    protected KcpRto rto;
    /**
     *  probe计算相关
     */
    protected KcpProbe probe;
    /**
     *  metric
     */
    protected KcpMetric metric;
    /**
     *  设置NoDelay参数
     */
    public int noDelay(boolean noDelay, int interval, int resend, boolean nc) {
        rto.setNoDelay(noDelay);
        if (noDelay) {
            rto.setRxMinrto(RpcKcpConfigInfo.INSTANCE.getiKcpRtoNdl());
        } else {
            rto.setRxMinrto(RpcKcpConfigInfo.INSTANCE.getiKcpRtoMin());
        }
        if (interval >= 0) {
            if (interval > 5000) {
                interval = 5000;
            } else if (interval < 10) {
                interval = 10;
            }
            this.interval = interval;
        }
        if (resend >= 0) {
            setFastResend(resend);
        }
        congestion.setNocwnd(nc);
        return 0;
    }
    /**
     * 设置窗口
     */
    public void wndSize(int sndWnd, int rcvWnd) {
        if (sndWnd > 0) { sender.setSndWnd(sndWnd); }
        if (rcvWnd > 0) { receiver.setRcvWnd(Math.max(rcvWnd, RpcKcpConfigInfo.INSTANCE.getiKcpWndRcv())); }
    }
    /**
     *  释放kcp
     */
    @Override
    public void release() {
        //  释放读写4个队列
        receiver.release();
        sender.release();
    }


    public boolean isAutoSetConv() {
        return autoSetConv;
    }

    public void setAutoSetConv(boolean autoSetConv) {
        this.autoSetConv = autoSetConv;
    }

    public int getConv() {
        return conv;
    }

    public void setConv(int conv) {
        this.conv = conv;
    }

    public int getCurrent() {
        return current;
    }

    public void setCurrent(int current) {
        this.current = current;
    }

    public KcpRto getRto() {
        return rto;
    }

    public void setRto(KcpRto rto) {
        this.rto = rto;
    }

    public KcpSender2 getSender() {
        return sender;
    }

    public void setSender(KcpSender2 sender) {
        this.sender = sender;
    }

    public KcpReceiver2 getReceiver() {
        return receiver;
    }

    public void setReceiver(KcpReceiver2 receiver) {
        this.receiver = receiver;
    }

    public int getRmtWnd() {
        return rmtWnd;
    }

    public void setRmtWnd(int rmtWnd) {
        this.rmtWnd = rmtWnd;
    }

    public AckList getAckList() {
        return ackList;
    }

    public void setAckList(AckList ackList) {
        this.ackList = ackList;
    }

    public int getMtu() {
        return mtu;
    }

    public void setMtu(int mtu) {
        this.mtu = mtu;
    }

    public int getMss() {
        return mss;
    }

    public void setMss(int mss) {
        this.mss = mss;
    }

    public int getState() {
        return state;
    }

    public void setState(int state) {
        this.state = state;
    }

    public boolean isUpdated() {
        return updated;
    }

    public void setUpdated(boolean updated) {
        this.updated = updated;
    }

    public KcpProbe getProbe() {
        return probe;
    }

    public void setProbe(KcpProbe probe) {
        this.probe = probe;
    }

    public KcpCongestion getCongestion() {
        return congestion;
    }

    public void setCongestion(KcpCongestion congestion) {
        this.congestion = congestion;
    }

    public int getFastResend() {
        return fastResend;
    }

    public void setFastResend(int fastResend) {
        this.fastResend = fastResend;
    }

    public KcpMetric getMetric() {
        return metric;
    }

    public void setMetric(KcpMetric metric) {
        this.metric = metric;
    }

    public int getFastLimit() {
        return fastLimit;
    }

    public void setFastLimit(int fastLimit) {
        this.fastLimit = fastLimit;
    }

    public int getDeadLink() {
        return deadLink;
    }

    public void setDeadLink(int deadLink) {
        this.deadLink = deadLink;
    }

    public int getInterval() {
        return interval;
    }

    public void setInterval(int interval) {
        this.interval = interval;
    }

    public boolean isStream() {
        return stream;
    }

    public void setStream(boolean stream) {
        this.stream = stream;
    }

    public ByteBufAllocator getByteBufAllocator() {
        return byteBufAllocator;
    }

    public void setByteBufAllocator(ByteBufAllocator byteBufAllocator) {
        this.byteBufAllocator = byteBufAllocator;
    }

    public Channel getChannel() {
        return channel;
    }

    public void setChannel(Channel channel) {
        this.channel = channel;
        kcpName = String.format("kcp{conv:%d, channel:%s}", conv, channel == null ? "NULL" : channel);
    }
    
    public void setKcpOutput(IKcpOutput output) {
    	getSender().setOutput(output);
    }
    
    protected String kcpName;
	@Override
	public String toString() {
		return kcpName;
	}
    
}
